package com.tang.consumer.netty;

import com.alibaba.fastjson.JSON;
import com.tang.consumer.connection.ConnectManage;
import com.tang.consumer.domain.Request;
import com.tang.consumer.domain.Response;
import com.tang.consumer.netty.codec.json.ProtocolMessage;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import lombok.extern.slf4j.Slf4j;

import java.nio.charset.Charset;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.SynchronousQueue;

/**
 * @author Tang
 * @classname NettyClientHandler
 * @description [  ]
 * @date 2021/1/24 20:17
 */
@Slf4j
public class NettyClientHandler extends ChannelInboundHandlerAdapter {

    private ConnectManage connectManage;

    public NettyClientHandler(ConnectManage connectManage) {
        this.connectManage = connectManage;
    }

    private final Map<String, SynchronousQueue<Object>> resultMap = new HashMap<>(16);

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        log.info("连接到RPC服务器:{}", ctx.channel().remoteAddress());
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        log.info("与RPC服务器断开连接:{}", ctx.channel().remoteAddress());
        connectManage.remove(ctx.channel());
        ctx.close();
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        if (msg instanceof ProtocolMessage) {
            // 解析包
            ProtocolMessage protocolMessage = (ProtocolMessage) msg;
            Response response = JSON.parseObject(new String(protocolMessage.getMessage()), Response.class);
            String requestId = response.getRequestId();
            // put 结果
            SynchronousQueue<Object> synchronousQueue = resultMap.get(requestId);
            synchronousQueue.put(response);
            // 删除这条记录
            resultMap.remove(requestId);
        }
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        log.info("已超过30秒未与RPC服务器进行读写操作!将发送心跳消息...");
        if (evt instanceof IdleStateEvent){
            IdleStateEvent event = (IdleStateEvent)evt;
            if (event.state()== IdleState.ALL_IDLE){
                // 发送心跳请求消息
                Request request = new Request();
                request.setMethodName("heartBeat");
                ProtocolMessage message = new ProtocolMessage();
                message.setMessage(JSON.toJSONString(request).getBytes());
                message.setLen(message.getMessage().length);
                ctx.channel().writeAndFlush(message);
            }
        }else{
            super.userEventTriggered(ctx,evt);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.info("RPC通信异常：{}",cause.getMessage());
        ctx.close();
    }

    /**
     * 发送请求
     * @param request
     * @param channel
     * @return
     */
    public SynchronousQueue<Object> sendRequest(Request request, Channel channel) {
        // 构造一个包
        ProtocolMessage message = new ProtocolMessage();
        message.setMessage(JSON.toJSONString(request).getBytes());
        message.setLen(message.getMessage().length);
        // 发送
        channel.writeAndFlush(message);
        SynchronousQueue<Object> synchronousQueue = new SynchronousQueue<>();
        // put 一个
        resultMap.put(request.getId(), synchronousQueue);
        return synchronousQueue;
    }
}